/*
 * BSM6.0
 *
 * Copyright 2012-2013 (C) SINODATA CO., LTD. All Rights Reserved.
 */
package com.sinodata.bsm.center.message;

import java.util.HashSet;
import java.util.Set;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.log4j.Logger;

import com.sinodata.bsm.center.engine.property.PropertyValueDistributeCenter;
import com.sinodata.bsm.common.message.BsmActiveMQFactory;
import com.sinodata.bsm.common.vo.ResPropertyValue;

/**
 * <p>
 * Description: 
 * </p>
 *
 * @author tangli
 * @version 1.0

 * <p>
 * History: 
 *
 * Date                     Author         Version     Description
 * ---------------------------------------------------------------------------------
 * 2012-6-12 下午5:36:10          tangli         1.0         To create
 * </p>
 *
 * @since 
 * @see     
 */
public class PropertyValueListener implements MessageListener {
    private static final Logger logger = Logger.getLogger(PropertyValueListener.class);
    private Session session;
    private Destination destination;
    private String queueName = null;
    private MessageConsumer consumer = null;
    private int ackMode = Session.AUTO_ACKNOWLEDGE;
    private static Set<Long> listenerSet = new HashSet<Long>();

    /**
     * 
     */
    private PropertyValueListener(Long probeId) {
        try {
            queueName = "PROP_" + probeId;
            session = BsmActiveMQFactory.getInstance().getSession(false, ackMode);
            destination = session.createQueue(queueName);
            consumer = session.createConsumer(destination);
            clean(consumer);
            consumer.setMessageListener(this);
        } catch (Exception ex) {
            logger.error("error create message queue:" + this.queueName, ex);
        }
    }

    /**
     * 
     * @param consumer
     */
    private void clean(MessageConsumer consumer) {
        while (true) {
            try {
                Message msg = consumer.receive(1000L);
                if (msg == null) {
                    logger.info("队列中没有消息");
                    break;
                }
            } catch (JMSException e) {
                logger.error("清空消息队列失败", e);
                break;
            }
        }
    }

    public static void newInstance(Long probeId) {
        if (listenerSet.contains(probeId)) {
            return;
        }
        new PropertyValueListener(probeId);
        listenerSet.add(probeId);
    }

    public void onMessage(Message message) {
        try {
            ObjectMessage objmsg = (ObjectMessage) message;
            Object msg = objmsg.getObject(); //待序列化的消息
            if (msg instanceof ResPropertyValue) {
                PropertyValueDistributeCenter.getInstance().dispose((ResPropertyValue) msg);
            }
        } catch (Exception e) {
            logger.error("failed to get method", e);
        }
    }
}
